package com.service.callback;

import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.parser.Feature;
import com.google.protobuf.InvalidProtocolBufferException;
import com.google.protobuf.util.JsonFormat;
import com.protos.Peer;
import org.apache.http.HttpResponse;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClientBuilder;
import org.apache.http.util.EntityUtils;
import org.java_websocket.client.WebSocketClient;
import org.java_websocket.handshake.ServerHandshake;

import java.net.URI;
import java.nio.ByteBuffer;
import java.util.Calendar;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.text.SimpleDateFormat;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import com.websocket.wsCallBack;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;


/**
 * 主要是用来查询交易是否上链
 *
 * @author jby
 */
public class CallbackOfTransactionOnChain extends WebSocketClient {

    private Logger logger = LoggerFactory.getLogger(this.getClass());
    private static Logger logger_static = LoggerFactory.getLogger(CallbackOfTransactionOnChain.class);

    private static CallbackOfTransactionOnChain push = null;
    private static int pullInitDelay = 10;//unit second
    private static int pullPeriod = 10;//unit second  如果是生产环境，可以扩大到15s

    private String host = "";

    private boolean isDebug = false;

    private ConcurrentHashMap<String, CallbackOfInner> waitingSubmitTxIds = null;
    private ExecutorService callbackThreadPool = Executors.newFixedThreadPool(20);//如果使用量大，改为30 或者 采用cachepool
    private ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor();

    private CallbackOfTransactionOnChain(String host) {
        super(URI.create("ws://" + host + "/event"));
        this.host = host;
        this.waitingSubmitTxIds = new ConcurrentHashMap<String, CallbackOfInner>();
    }

    /**
     * 管理线程，主要是超时管理
     * 类似于状态收集器
     */
    private void startupMgrThread() {
        this.scheduledExecutorService.scheduleAtFixedRate(() -> {
            try {
                if (!waitingSubmitTxIds.isEmpty()) {
                    List<String> tmpls = new ArrayList<String>();
                    for (String key : waitingSubmitTxIds.keySet()) {
                        try {
                            CallbackOfInner ci = waitingSubmitTxIds.get(key);
                            if (ci == null) {
                                //wait delete
                                tmpls.add(key);
                            } else {
                                if (ci.callbackStatus()) {
                                    //callback and delete
                                    startupCallback(ci);
                                    tmpls.add(key);
                                } else {
                                    if (ci.isTimeout()) {
                                        //timeout failed callback and delete
                                        startupCallback(ci);
                                        tmpls.add(key);
                                    } else {
                                        if (ci.isPullTimeout()) {
                                            ci.setPullLastTime();
                                            //start pull thread
                                            startupPull(ci);
                                        }
                                    }
                                }
                            }
                        } catch (Exception e) {
                            logger.error("CallbackOfTransactionOnChain:scheduleAtFixedRate, loop error ,error msg= {}", e.getMessage());
                        }
                    }

                    for (String k : tmpls) {
                        try {
                            waitingSubmitTxIds.remove(k);
                        } catch (Exception el) {
                            logger.error("CallbackOfTransactionOnChain:scheduleAtFixedRate, remove key error ,error msg={}", el.getMessage());
                        }
                    }
                }
                if (isDebug) {
                    logger.debug("CallbackOfTransactionOnChain:scheduleAtFixedRate, info ,map size={}", waitingSubmitTxIds.size());
                }
                isReady();
            } catch (Exception em) {
                logger.error("CallbackOfTransactionOnChain:scheduleAtFixedRate, thread run error ,error msg={}", em.getMessage());
            }

        }, pullInitDelay, pullPeriod, TimeUnit.SECONDS);
    }

    /**
     * 单例模式的实例获取
     *
     * @param host
     * @return
     */
    public static synchronized CallbackOfTransactionOnChain getPush(String host) {
        if (push == null) {
            push = new CallbackOfTransactionOnChain(host);
            try {
                push.connectBlocking();
                push.startupMgrThread();
            } catch (InterruptedException e) {
                logger_static.error("CallbackOfTransactionOnChain:{}, opened failed ,error msg={} ", push.getURI().getHost(), e.getMessage());
            }
        }
        return push;
    }

    /**
     * 添加 txid::wsCallBack 对
     *
     * @param txid
     * @param wc
     */
    public void addCallbackInfo(String txid, wsCallBack wc) {
        //todo add callback
        CallbackOfInner callback = new CallbackOfInner(wc, txid);
        this.waitingSubmitTxIds.put(txid, callback);
    }

    public void removeCallbackInfo(String txid) {
        this.waitingSubmitTxIds.remove(txid);
    }

    //判断websocet的状态，并主动发起连接
    private boolean isReady() {
        if (this.isOpen()) {
            return true;
        } else if (this.isClosed()) {
            try {
                this.reconnectBlocking();
                if (this.isDebug)
                    logger.debug("CallbackOfTransactionOnChain:isReady, connectBlocking");
            } catch (InterruptedException e) {
                logger.error("CallbackOfTransactionOnChain: {}, opened failed ,error msg={}", this.getURI().getHost(), e.getMessage());
                return false;
            }
            return true;
        } else if (this.isClosing() || this.isConnecting()) {
            try {
                this.reconnectBlocking();
                if (this.isDebug)
                    logger.debug("CallbackOfTransactionOnChain:isReady, reconnectBlocking");
            } catch (InterruptedException e) {
                logger.error("CallbackOfTransactionOnChain:{}, opened failed ,error msg={}", this.getURI().getHost(), e.getMessage());
                return false;
            }
            return true;
        } else {
            return false;
        }
    }

    @Override
    public void onOpen(ServerHandshake handshakedata) {
        logger.info("CallbackOfTransactionOnChain:{}, opened connection ,handshake={}", this.getURI().getHost(), handshakedata.getHttpStatusMessage());
    }

    @Override
    public void onMessage(String message) {
        logger.info("CallbackOfTransactionOnChain:{}, received message is {}", this.getURI().getHost(), message);
    }

    @Override
    public void onClose(int code, String reason, boolean remote) {
        logger.info("CallbackOfTransactionOnChain:{}, Connection closed，code={},reason={}，isRemote={}", remote, this.getURI().getHost(), code, reason, remote);
    }

    @Override
    public void onError(Exception ex) {
        logger.error("CallbackOfTransactionOnChain:{}, an error occurred:{}", this.getURI().getHost(), ex.getLocalizedMessage());
    }

    @Override
    public void onMessage(ByteBuffer message) {
        try {
            byte[] bytes = new byte[message.capacity()];
            message.get(bytes, 0, bytes.length);
            Peer.Event event = Peer.Event.parseFrom(bytes);
            if (event != null && event.hasBlk()) {
                //todo check transaction to store
                Peer.Block block = event.getBlk();
                for (Peer.Transaction tran : block.getTransactionsList()) {
                    //只选择本实例中需要回调的交易进行操作，在等待map中如果不存在就不执行以下工作
                    if (this.waitingSubmitTxIds.containsKey(tran.getId())) {
                        CallbackOfInner ci = this.waitingSubmitTxIds.get(tran.getId());
                        //上链成功
                        ci.onChainSuccess();
                        //进入回调，开启回调线程进行回调
                        startupCallback(ci);
                        if (this.isDebug)
                            logger.debug("onmessage:txid={}", ci.getTxid());
                    }
                }
            }
        } catch (Exception ex) {
            logger.error("CallbackOfTransactionOnChain:{} ,onMessage4ByteBuffer error occurred:{}", this.getURI().getHost(), ex.getLocalizedMessage());
        }
    }

    private static String getDateStr() {
        String str = "";
        String formatstr = "yyyy-MM-dd HH:mm:ss";
        try {
            SimpleDateFormat sdf = new SimpleDateFormat(formatstr);
            str = sdf.format(Calendar.getInstance().getTime());
        } catch (Exception e) {
            e.printStackTrace();
        }
        return str;
    }

    //添加pull到固定线程池，然后再去运行，大于线程池大小，线程将等待运行
    private void startupPull(CallbackOfInner ci) {
        if (ci != null) {
            PullThread ct = new PullThread(ci);
            this.callbackThreadPool.execute(ct);
        }
    }

    //添加回调到固定线程池，然后再去运行，大于线程池大小，线程将等待运行
    // 改线程主要是做回调操作，不论成功与否
    private void startupCallback(CallbackOfInner ci) {
        if (ci != null) {
            this.waitingSubmitTxIds.remove(ci.getTxid());
            CallbackThread ct = new CallbackThread(ci);
            this.callbackThreadPool.execute(ct);
        }
    }

    public class CallbackThread implements Runnable {

        private CallbackOfInner ci = null;

        public CallbackThread(CallbackOfInner ci) {
            this.ci = ci;
        }

        @Override
        public void run() {
            try {
                this.ci.onCallback();
            } catch (Exception e) {
                logger.error("CallbackOfTransactionOnChain:CallbackThread  failed ,txid={},error msg={}", this.ci.getTxid(), e.getMessage());
            }
        }
    }

    /**
     * 拉取线程
     */
    public class PullThread implements Runnable {

        private CallbackOfInner ci = null;

        public PullThread(CallbackOfInner ci) {
            this.ci = ci;
        }

        @Override
        public void run() {
            try {
                Peer.Transaction t = getTransactionByTxid(ci.getTxid());
                if (t != null) {
                    ci.onChainSuccess();
                    //进入回调，开启回调线程进行回调
                    startupCallback(ci);
                    if (isDebug)
                        logger.debug("PullThread:txid={}", ci.getTxid());
                }
            } catch (Exception e) {
                logger.error("CallbackOfTransactionOnChain:CallbackThread  failed ,txid={},error msg={}", this.ci.getTxid(), e.getMessage());
            }
        }

        /**
         * 查询交易是否上链
         *
         * @param txid
         * @return
         */
        private Peer.Transaction getTransactionByTxid(String txid) {
            JSONObject jsonObject = doGet("http://" + host + "/transaction/" + txid);
            JSONObject result = jsonObject.getJSONObject("result");
            if (result == null || result.isEmpty()) {
                return null;
            }
            Peer.Transaction.Builder builder = Peer.Transaction.newBuilder();
            try {
                String json = result.toJSONString();
                JsonFormat.parser().merge(json, builder);
            } catch (InvalidProtocolBufferException e) {
                e.printStackTrace();
            }
            return builder.build();
        }

        private JSONObject doGet(String url) {
            CloseableHttpClient client = HttpClientBuilder.create().build();
            HttpGet get = new HttpGet(url);
            JSONObject jsonObject = new JSONObject(true);
            try {
                HttpResponse result = client.execute(get);
                String str = EntityUtils.toString(result.getEntity());
                jsonObject = (JSONObject) JSONObject.parse(str, Feature.OrderedField);
            } catch (Exception ex) {
                ex.printStackTrace();
            } finally {
                if (client != null) {
                    try {
                        client.close();
                    } catch (Exception ex) {
                        ex.printStackTrace();
                    }
                }
            }
            return jsonObject;
        }
    }

    /**
     * 测试专用接口，非测试环境不要调用
     */
    public void setDebug() {
        this.isDebug = true;
    }

    public void setConnectionClose() {
        this.close();
    }
}
